[AWS IoT SiteWise] disassociated modeについて確認してみました

SIN

2024.01.23

[AWS IoT SiteWise]

1 はじめに

CX事業本部製造ビジネステクノロジー部の平内（SIN）です。

AWS IoT SiteWise のドキュメントの「データ取り込みを設定」では、disassociated-data-storageについて記述があるのでが、ちょっと上手く理解できなかったので、少し試してみました。

https://docs.aws.amazon.com/ja_jp/iot-sitewise/latest/userguide/manage-data-streams.html#manage-data-streams-prerequisites

2 disassociatedDataStorage

AWS IoT SiteWise ストレージの現在のdisassociatedDataStorage の設定値については、describe-storage-configurationで確認できます。

% aws iotsitewise describe-storage-configuration
{
    "storageType": "SITEWISE_DEFAULT_STORAGE",
    "disassociatedDataStorage": "ENABLED",
    "configurationStatus": {
        "state": "ACTIVE"
    }
}

disassociatedDataStorageの設定値には、DISABLED及び、ENABLEDがありますが、それぞれの動作は以下のとおりです。

disassociatedDataStorage	モード名	動作	備考
DISABLED	associated mode	アセットプロパティに関連付けられたデータストリームのみデータストアに受けいれる
ENABLED	disassociated mode	アセットプロパティに関連付けに関係なくすべてのデータストリームをデータストアに受けいれる	デフォルト値

２種類の設定値があるのですが、2021/11/24以降、デフォルト値がENABLEDになっていて、DISABLEDには変更できないようです。

% aws iotsitewise put-storage-configuration \
    --storage-type SITEWISE_DEFAULT_STORAGE \
    --disassociated-data-storage DISABLED

An error occurred (InvalidRequestException) when calling the PutStorageConfiguration operation: The only accepted value for DisassociatedDataStorage field is ENABLED

また、ドキュメントに記載されている、[設定]の階層下の[データ取り込み]は、そのメニューが表示されていませんでした。

3 disassociated mode

disassociatedDataStorageが、ENABLEDとなっている場合、アセットプロパティに紐づかないデータも、すべてSiteWiseの時系列データベースに保存されるということなので、その状況を確認してみました。

下記は、batch_put_asset_property_valueを使用して、エリアス指定でSiteWiseにデータを送信しているプログラムです。

batch_put_asset_property_valueでは、アセットIDとプロパティIDを指定して、アセットに紐づいたデータを送信することもできますが、ここでは、エリアスだけを指定しています。

1秒毎に3回だけランダムな数値を送信しています。

import boto3
from uuid import uuid4
import random
import time
from datetime import datetime


def put_asset(client, property_alias, value):
    client.batch_put_asset_property_value(
        entries=[
            {
                "entryId": str(uuid4()),
                "propertyAlias": property_alias,
                "propertyValues": [
                    {
                        "value": {"doubleValue": value},
                        "timestamp": {
                            "timeInSeconds": int(datetime.now().timestamp()),
                            "offsetInNanos": 0,
                        },
                        "quality": "GOOD",
                    }
                ],
            }
        ]
    )


def main():
    client = boto3.client("iotsitewise")

    counter = 3
    wait_sec = 1

    for _ in range(counter):
        put_asset(client, "Sample_1", random.uniform(0, 100))
        put_asset(client, "Sample_2", random.uniform(0, 100))

        time.sleep(wait_sec)


if __name__ == "__main__":
    main()

上記プログラムを実行したあと、コンソールのデータストリームでは、このデータの最終更新日を確認できます。

また、list_time_seriesでは、timeSeriesTypeが、DISASSOCIATEDとなったデータとして列挙され、get_asset_property_value_historyで、時系列のデータを取得できます。

import boto3
import datetime

client = boto3.client("iotsitewise")


def list_time_series(timeSeriesType):
    max_results = 250
    summaries = []
    key = "TimeSeriesSummaries"
    response = client.list_time_series(
        maxResults=max_results, timeSeriesType=timeSeriesType
    )
    if key in response:
        summaries = summaries + response[key]

    while True:
        if not "nextToken" in response:
            break
        response = client.list_time_series(
            nextToken=response["nextToken"],
            maxResults=max_results,
            timeSeriesType=timeSeriesType,
        )
        if key in response:
            summaries = summaries + response[key]

    return summaries


def get_asset_property_value_history(alias):
    history = []
    max_results = 1000
    key = "assetPropertyValueHistory"
    response = client.get_asset_property_value_history(
        maxResults=max_results, propertyAlias=alias
    )

    if key in response:
        history = history + response[key]

    while True:
        if not "nextToken" in response:
            break
        response = client.get_asset_property_value_history(
            nextToken=response["nextToken"], maxResults=max_results, propertyAlias=alias
        )

        if key in response:
            history = history + response[key]

    return history


def main():
    summaries = list_time_series("DISASSOCIATED")
    for summarie in summaries:
        if "alias" in summarie:
            if summarie["alias"].startswith("Sample_"):
                alias = summarie["alias"]
                print("alias: {}".format(alias))

                history = get_asset_property_value_history(alias)
                for data in history:
                    timeInSeconds = data["timestamp"]["timeInSeconds"]
                    timestamp = datetime.datetime.fromtimestamp(timeInSeconds)
                    value = data["value"]
                    print("{} {}".format(timestamp, value))


if __name__ == "__main__":
    main()

% python3 index.py
alias: Sample_1
2024-01-23 01:25:41 {'doubleValue': 74.37977791887853}
2024-01-23 01:25:43 {'doubleValue': 55.1891373855083}
2024-01-23 01:25:44 {'doubleValue': 54.78059871389871}
alias: Sample_2
2024-01-23 01:25:42 {'doubleValue': 82.5442431812442}
2024-01-23 01:25:43 {'doubleValue': 17.590713341138}
2024-01-23 01:25:44 {'doubleValue': 29.010846096747088}